package ws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.util.HashedWheelTimer;
import netty.util.StartAndStop;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.jctools.maps.NonBlockingHashSet;
import test.PerformanceAnalyze;
import ws.client.WebSocketClient;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class OkHttpWsPrivate extends StartAndStop {

    static PerformanceAnalyze performanceAnalyze = new PerformanceAnalyze();

    private void performanceAnalyze(JSONObject jsonObject) {
        long creationTime = jsonObject.getLongValue("creationTime");
        if (creationTime != 0) {
            performanceAnalyze.delay(System.currentTimeMillis() - creationTime);
        }
    }

    public void start(String url, int size, int sleep) throws InterruptedException {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(50);
        HashedWheelTimer hashedWheelTimer =
                new HashedWheelTimer(
                        new NamedThreadFactory("LingerMessageSendTimer", false),
                        1, TimeUnit.MILLISECONDS,
                        1000, true);

        executor.scheduleAtFixedRate(() -> {
            NonBlockingHashSet<WebSocketClient> wsList = new NonBlockingHashSet<>();

            CountDownLatch clientCounter = new CountDownLatch(size);

            for (int i = 0; i < size; i++) {
                executor.execute(() -> {
                    try {
                        WebSocketClient webSocketClient = WebSocketClient.create(true, url, msg -> {

                            if (StringUtils.contains(msg, "topic")) {
                                sleep(sleep);
                            }

                            JSONObject jsonObject = JSON.parseObject(msg);
                            performanceAnalyze(jsonObject);

                            performanceAnalyze.length(msg.length());
                            performanceAnalyze.meter();
                        });

                        wsList.add(webSocketClient);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        clientCounter.countDown();
                    }
                });
            }

            try {
                clientCounter.await(20, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            for (WebSocketClient webSocket : wsList) {
                // {"id":"1648560477000","op":"login","args":["1"]}
                Map<String, Object> login = new HashMap<>();
                login.put("id", String.valueOf(System.currentTimeMillis()));
                login.put("op", "login");
                List<String> loginArgs = new LinkedList<>();
                loginArgs.add(String.valueOf(RandomUtils.nextInt(800300, 800899)));
                login.put("args", loginArgs);
                String loginCommand = JSON.toJSONString(login);
                webSocket.send(loginCommand);

                // {"op":"subscribe","id":"1645784808000","args":["1"]}
                // {"op":"fuzzySub","id":"1648539957000","args":["ticker"]}
                Map<String, Object> subscribe = new HashMap<>();
                subscribe.put("id", String.valueOf(System.currentTimeMillis()));
                subscribe.put("op", "subscribe");
                List<String> args = new LinkedList<>();
                args.add("user.order");
                subscribe.put("args", args);
                String command = JSON.toJSONString(subscribe);
                webSocket.send(command);
            }

            for (WebSocketClient webSocket : wsList) {
                Map<String, Object> ping = new HashMap<>();
                ping.put("id", String.valueOf(System.currentTimeMillis()));
                ping.put("op", "ping");

//                executor.scheduleAtFixedRate(() -> webSocket.send(JSON.toJSONString(ping)), 10, 10, TimeUnit.SECONDS);
//                hashedWheelTimer.newTimeout((timeout) -> webSocket.close(), 10, TimeUnit.SECONDS);
            }
        }, 0, 30, TimeUnit.SECONDS);
    }


    private void sleep(int sleep) {
        try {
            Thread.sleep(sleep);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
